package video

import (
	"encoding/json"
	"fmt"
	"strings"
	"time"
)

const CacheKey = "video_watch"

func WriteCache(uid, vid int64, task string, progress int) error {
	// 从redis中读取缓存的数据
	rec := new(Record)
	cacheField := fmt.Sprintf("u:%d v:%d t:%s", uid, vid, task)
	err := r.HGet(CacheKey, cacheField).Scan(rec)
	needCreate := false // 要在数据库中创建新的记录
	// 如果找不到缓存的记录,或者不匹配，尝试从数据库中获取
	if err != nil || rec.StartProgress > progress+MaxReportInterval || rec.EndProgress < progress-MaxReportInterval {
		// 查找一个包含当前进度，或者接近当前进度的记录
		cond := "user_id = ? AND video_id = ? AND task_code = ? AND start_progress < ? AND end_progress > ?"
		argv := []interface{}{uid, vid, task, progress+10, progress-10}
		ok, err := sx.Where(cond, argv...).Get(rec)
		if err != nil {
			return err
		}
		needCreate = !ok
	}
	now := time.Now().Unix()
	if needCreate {
		rec = &Record{
			UserID:        uid,
			VideoID:       vid,
			TaskCode:      task,
			StartProgress: progress,
			EndProgress:   progress,
			LastProgress: progress,
			UpdateTime:    now,
		}
		_, err = sx.Insert(rec)
	} else {
		if rec.StartProgress > progress {
			rec.StartProgress = progress
		} else if rec.EndProgress < progress {
			rec.EndProgress = progress
		}
		if progress > rec.LastProgress {
			rec.WatchDuration += progress - rec.LastProgress
		}
		rec.LastProgress = progress
		rec.UpdateTime = now
	}
	if _, err = r.HSet(CacheKey, cacheField, rec).Result(); err != nil {
		return err
	}
	return nil
}

// 定时将redis中的数据同步到数据库
func StartCacheSync() {
	var lastTime int64 = 0 // 上次更新时间，只更新此时间之后更新的数据
	for {
		time.Sleep(time.Second * time.Duration(MaxReportInterval))
		l.Println("开始同步视频观看进度")
		now := time.Now().Unix()
		releaseTime := now - int64(MaxReportInterval) // 释放早于此时间的缓存
		data, err := r.HGetAll(CacheKey).Result()
		if err != nil {
			l.Println("错误:", err)
			continue
		}
		release := make([]string, 0) // 要释放的缓存的field
		values := make([]string, 0)
		for field, value := range data {
			rec := new(Record)
			if err := json.Unmarshal([]byte(value), rec); err == nil {
				if rec.UpdateTime < releaseTime {
					release = append(release, field)
				}
				if rec.UpdateTime > lastTime {
					values = append(values, fmt.Sprintf("(%d, %d, %d, '%s', %d, %d, %d, %d, %d)",
						rec.ID, rec.UserID, rec.VideoID, rec.TaskCode, rec.StartProgress, rec.EndProgress, rec.LastProgress, rec.WatchDuration, rec.UpdateTime))
				}
			}
		}
		if len(values) > 0 {
			query := "REPLACE INTO t_video_record(id, user_id, video_id, task_code, start_progress, end_progress, last_progress, watch_duration, update_time) VALUES " +
				strings.Join(values, ",")
			if _, err := sx.Exec(query); err != nil {
				l.Println("错误:", err)
				continue
			}
		}
		if len(release) > 0 {
			_, err = r.HDel(CacheKey, release...).Result()
			if err != nil {
				l.Println("错误:", err)
				continue
			}
		}
		l.Println("读取到缓存:", len(data), "更新记录:", len(values), "删除过期缓存:", len(release))
		lastTime = now
	}
}

/*
定时将统计观看记录中的进程并写入统计表
根据是否区分task_code分两种情况，都要进行统计
 */
func StartProgressStat()  {
	// 上次更新时间，只更新此时间之后更新的数据
	var lastTime int64 = 0
	if _, err := sx.Table("t_video_stat").OrderBy("update_time DESC").Cols("update_time").Get(&lastTime); err != nil {
		l.Println("错误:", err)
		return
	}
	for {
		l.Println("开始统计视频观看进度")
		now := time.Now().Unix()
		sm := make(map[string]Stat)
		em := make(map[string]int) // 前一个记录最后的进度
		lm := make(map[string]int64) // 最后一次更新时间
		{
			// 不区分task_code的统计
			query := "SELECT h.* FROM t_video_record AS r JOIN t_video_record AS h USING(user_id, video_id) " +
				"WHERE r.update_time >= ? ORDER BY h.start_progress"
			data := make([]Record, 0)
			if err := sx.SQL(query, lastTime).Find(&data); err != nil {
				l.Println("错误:", err)
				continue
			}
			for _, item := range data {
				key := fmt.Sprintf("u:%d v:%d", item.UserID, item.VideoID)
				stat, ok := sm[key]
				if !ok {
					stat = Stat{
						UserID:        item.UserID,
						VideoID:       item.VideoID,
					}
				}
				if e := em[key]; e < item.StartProgress {
					stat.ValidDuration += item.EndProgress - item.StartProgress
					em[key] = item.EndProgress
				} else if e < item.EndProgress {
					stat.ValidDuration += item.EndProgress - e
					em[key] = item.EndProgress
				}
				stat.TotalDuration += item.WatchDuration
				if item.UpdateTime > lm[key] {
					item.UpdateTime = lm[key]
					stat.LastProgress = item.LastProgress
				}
				sm[key] = stat
			}
		}
		{
			// 区分task_code，只统计task_code不为空的
			query := "SELECT h.* FROM t_video_record AS r JOIN t_video_record AS h USING(user_id, video_id, task_code) " +
				"WHERE r.update_time >= ? AND r.task_code != '' ORDER BY h.start_progress"
			data := make([]Record, 0)
			if err := sx.SQL(query, lastTime).Find(&data); err != nil {
				l.Println("错误:", err)
				continue
			}
			for _, item := range data {
				key := fmt.Sprintf("u:%d v:%d t:%s", item.UserID, item.VideoID, item.TaskCode)
				stat, ok := sm[key]
				if !ok {
					stat = Stat{
						UserID:        item.UserID,
						VideoID:       item.VideoID,
						TaskCode:      item.TaskCode,
					}
				}
				if e := em[key]; e < item.StartProgress {
					stat.ValidDuration += item.EndProgress - item.StartProgress
					em[key] = item.EndProgress
				} else if e < item.EndProgress {
					stat.ValidDuration += item.EndProgress - e
					em[key] = item.EndProgress
				}
				stat.TotalDuration += item.WatchDuration
				if item.UpdateTime > lm[key] {
					item.UpdateTime = lm[key]
					stat.LastProgress = item.LastProgress
				}
				sm[key] = stat
			}
		}
		// 将数据写入数据库
		if len(sm) > 0 {
			// 拼接视频时长
			vis := make([]int64, 0)
			for _, stat := range sm {
				vis = append(vis, stat.VideoID)
			}
			dm, err := MatchVideoDuration(vis)
			if err != nil {
				l.Println("错误:", err)
				continue
			}
			for key, stat := range sm {
				stat.VideoDuration = dm[stat.VideoID]
				sm[key] = stat
			}
			// 拼接sql
			values := make([]string, 0)
			for _, stat := range sm {
				values = append(values, fmt.Sprintf("(%d, %d, %d, '%s', %d, %d, %d, %d, %d)",
					stat.ID, stat.UserID, stat.VideoID, stat.TaskCode, stat.LastProgress, stat.ValidDuration, stat.TotalDuration, stat.VideoDuration, now))
			}
			query := "REPLACE INTO t_video_stat(id, user_id, video_id, task_code, last_progress, valid_duration, total_duration, video_duration, update_time) VALUES " +
				strings.Join(values, ",")
			if _, err := sx.Exec(query); err != nil {
				l.Println("错误:", err)
				continue
			}
		}
		lastTime = now
		time.Sleep(time.Second * time.Duration(ProgressStatInterval))
	}
}
